/**
 * 
 */
package DBside;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;

import datatypes.Datatypes.tableType;
import datatypes.tv.*;
import datatypes.film.*;

import Utils.DebugUtils;
import Utils.LoadingCenter;
import Utils.Concurrent.TaskRunError;
import Utils.Freebase.FB_ParsingUtils;

import interfaces.IDataBaseOperation;

/**
 * @author Ortal & Yosi & Micha & Yuval
 * 
 */
	public class DataBaseOperation implements IDataBaseOperation {
		private Connection conn;
		private static final int executionValue = 5000;

		@Override
		public void createDataBase(Connection conn) throws TaskRunError {

			this.conn = conn;

			// load data types

			// Get a Runtime object
			Runtime r = Runtime.getRuntime();

			DebugUtils.initTimer();
			DebugUtils.startTimer();

			// Collect garbage at the start of the program

			/* --- tv parsing --- */

			tvActors tv_actors = (tvActors) LoadingCenter
					.loadType(tableType.tvActor);

			System.out.println("tv actors size : "+tv_actors.getActors().size());
			tvPrograms tv_programs = (tvPrograms) LoadingCenter
					.loadType(tableType.tvProgram);
			System.out.println("tv program size : "+tv_programs.getPrograms().size());

			tvProgramCreators tv_program_creators = (tvProgramCreators) LoadingCenter
					.loadType(tableType.tvProgramCreator);
			System.out.println("tv program creators : "+tv_program_creators.getProgramCreators().size());

			tvStarringRoles starringRoles1 = (tvStarringRoles) LoadingCenter
					.loadType(tableType.tvStarringRole);
			System.out.println("tv starring role : "+starringRoles1.gettvStarringRoles().size());

			Map<String, Integer> tvCreatorsMap = new HashMap<String, Integer>();
			Map<String, tvStarringRole> tvStarringRoles = new HashMap<String, tvStarringRole>();
			List<tvProgramCreator> programCreators;
			List<CreatedTable> createdTableList = new ArrayList<CreatedTable>();
			int id = 0;

			List<tvStarringRole> tvStarringRolesList;
			// get actor id's and starring roles
			for (tvActor actor : tv_actors.getActors()) {
				id++;
				tvStarringRolesList = actor.getActorStarringRoles()
						.gettvStarringRoles();
				if (!tvStarringRolesList.isEmpty()) {
					for (tvStarringRole role : tvStarringRolesList) {
						if (role.getId() != null) {
							role.setActorId(id);
							tvStarringRoles.put(role.getId(), role);
						}
					}
				}
			}
			// get tv program creator
			id = 0;

			for (tvProgramCreator creator : tv_program_creators
					.getProgramCreators()) {
				id++;
				tvCreatorsMap.put(creator.getId(), id);
			}

			// get tv program
			id = 0;
			for (tvProgram program : tv_programs.getPrograms()) {
				id++;
				program.setProgramId(id);
				programCreators = program.getProgram_creators()
						.getProgramCreators();
				if (!programCreators.isEmpty()) {
					for (tvProgramCreator creator : programCreators) {
						if (creator.getId() != null
								&& tvCreatorsMap.get(creator.getId()) != null) {
							CreatedTable createdTable = new CreatedTable(id,
									tvCreatorsMap.get(creator.getId()));
							createdTableList.add(createdTable);
						}

					}
				}
				tvStarringRolesList = program.getRegular_cast()
						.gettvStarringRoles();
				if (!tvStarringRolesList.isEmpty()) {
					for (tvStarringRole role : tvStarringRolesList) {
						tvStarringRole tr = tvStarringRoles.get(role.getId());
						if (tr != null) {
							tr.setProgramId(id);
							tvStarringRoles.put(role.getId(), tr);
						}
					}
				}
			}

			 insertActorBatch(tv_actors);
			 insertProgram(tv_programs);
			 insertRegularAppearence(tvStarringRoles);
			 insertProgramCreator(tv_program_creators);
			 insertCreatedTable(createdTableList);

			tv_actors = null;
			tv_programs = null;
			tvStarringRoles = null;
			tv_program_creators = null;
			tvCreatorsMap = null;
			tvStarringRoles = null;
			starringRoles1 = null;
			programCreators = null;
			createdTableList = null;
			FB_ParsingUtils.starringRoleMap = null;
			FB_ParsingUtils.programCreatorMap = null;
			System.gc();
			r.gc();
			// get film director
			id = 0;

			List<filmRole> filmRoleList = new ArrayList<filmRole>();
			Map<String, filmRole> filmRoleMap = new TreeMap<String, filmRole>();

			// parse directors
			Map<String, Integer> directorsMap = new HashMap<String, Integer>();
			filmDirectors film_director = (filmDirectors) LoadingCenter
					.loadType(tableType.filmDirector);
			System.out.println("film director size : "+film_director.getfilmDirectors().size());
			
			id=0;
			for (filmDirector director : film_director.getfilmDirectors()) {
				id++;
				directorsMap.put(director.getName(), id);
			}

			// insert to database and release memory
			 insertDirectorBatch(film_director);
			film_director = null;

			/* --- film parsing --- */

			List<DirectorTable> directorTableList = new ArrayList<DirectorTable>();
			Films films = (Films) LoadingCenter.loadType(tableType.film);
			 
			System.out.println("films size : "+films.getFilms().size());

			 handleFilms(films, directorTableList, directorsMap, filmRoleList,filmRoleMap);
			films = null;

			// insert directs table and release memory
			insertDirectorTableBatch(directorTableList);
			directorTableList = null;
			directorsMap = null;
			System.gc();

			// parse film actors

			filmActors film_actors = (filmActors) LoadingCenter
					.loadType(tableType.filmActor);
			System.out.println("film actor size : "+film_actors.getFilmActors().size());
			
			id = 0;
			String[] starringRoles3;
			for (filmActor filmActor : film_actors.getFilmActors()) {
				id++;
				filmActor.setFilmActorId(id);
				starringRoles3 = filmActor.getFilmRoleMids().split(",");
				if (starringRoles3.length > 0) {
					for (String role : starringRoles3) {
						if (role != null) {

							filmRole fr = filmRoleMap.get(role);
							if (fr != null) {
								fr.setFilmActorId(id);
							}
						}
					}
				}
			}

			System.out.println("before film actor batch");
			// insert film actors to the table and release memory
			 insertFilmActorBatch(film_actors);
			film_actors = null;

			// insert film roles and release memory
			System.out.println("role size : " + filmRoleMap.size());
			 insertFilmStarringRole(filmRoleMap);
			filmRoleMap = null;

			System.gc();

			filmRoles film_roles = (filmRoles) LoadingCenter
					.loadType(tableType.filmRole);
			System.out.println("film role size : "+film_roles.getFilm_roles().size());

			updateCharacterFilmPlayBatch(film_roles.getFilm_roles());
			film_roles = null;

			System.out.println(DebugUtils.TimeToString(DebugUtils.stopTimer()));
			
			this.conn = null;
		}

		private void handleFilms(Films films,
				List<DirectorTable> directorTableList,
				Map<String, Integer> directorsMap, List<filmRole> filmRoleList,
				Map<String, filmRole> filmRoleMap) throws TaskRunError {
			// get films
			int id = 0;
			String[] directorsList;
			String[] starringRoles;
			for (Film film : films.getFilms()) {
				id++;
				film.setFilmId(id);
				directorsList = film.getDirectorMids().split(",");
				if (directorsList.length > 0) {
					for (String director : directorsList) {
							if (directorsMap.get(director) != null) {
								DirectorTable directorTable = new DirectorTable(
										directorsMap.get(director), id);
								directorTableList.add(directorTable);
							}
						
					}
				}

				starringRoles = film.getStarringMids().split(",");

				if (starringRoles.length > 0) {
					for (String role : starringRoles) {
						if (role != null) {
							filmRole fr = new filmRole();
							fr.setFilmId(id);
							fr.setId(role);
							filmRoleMap.put(role, fr);
							// filmRoleList.add(fr);
						}
					}
				}
			}

			// insert film into database and release memory
			insertFilmBatch(films);

		}

		private void setAutoIncremet(String tableName, int value)
				throws TaskRunError {
			Statement st = null;
			try {
				st = conn.createStatement();
			} catch (SQLException e1) {
				throw new TaskRunError("setAutoIncremet connection error "
						+ tableName);
			}
			String sql = "ALTER TABLE " + tableName + " auto_increment="
					+ value;
			try {
				st.executeUpdate(sql);
			} catch (SQLException e) {
				throw new TaskRunError("setAutoIncremet executeUpdate error "
						+ tableName);
			}
		}

		private filmRole getRoleByID(List<filmRole> roleLst, String filmRoleID) {
			for (filmRole film_role : roleLst) {
				if (film_role.getId().equals(filmRoleID)) {
					return film_role;
				}
			}
			return null;

		}

		@Override
		public void removeDataBase(Connection conn) throws TaskRunError {
			this.conn = conn;

			
			deleteTable("Created");
			setAutoIncremet("Created", 1);
			deleteTable("Directs");
			setAutoIncremet("Directs", 1);
			deleteTable("Film_Play");
			setAutoIncremet("Film_Play", 1);
			deleteTable("TV_Play");
			setAutoIncremet("TV_Play", 1);
			
			deleteTable("Director");
			setAutoIncremet("Director", 1);
			deleteTable("Film");
			setAutoIncremet("Film", 1);
			deleteTable("Film_Actor");
			setAutoIncremet("Film_Actor", 1);
			deleteTable("Program_Creator");
			setAutoIncremet("Program_Creator", 1);
			deleteTable("TV_Actor");
			setAutoIncremet("TV_Actor", 1);
			deleteTable("TV_Program");
			setAutoIncremet("TV_Program", 1);

		//	deleteTable("Users");
		//	setAutoIncremet("Users", 1);
			
			this.conn = null;
		}

		private void insertFilmStarringRole(
				Map<String, filmRole> filmStarringRole) throws TaskRunError {
			PreparedStatement pstmt;
			try {
				pstmt = conn
						.prepareStatement("INSERT INTO Film_Play(Actor_Id,Film_Id,`Character`,mid) "
								+ "VALUES(?,?,?,?)");
				conn.setAutoCommit(false);
				Collection<filmRole> values = filmStarringRole.values();

				int batchCounter = 0;

				for (filmRole role : values) {
					if ((role.getFilmActorId() != -1)
							&& (role.getFilmId() != -1)) {

						if (role.getId().length() == 0) {
							batchCounter++;
							continue;
						}

						pstmt.setInt(1, role.getFilmActorId());
						pstmt.setInt(2, role.getFilmId());
						pstmt.setString(3, "");
						pstmt.setString(4, role.getId());
						pstmt.addBatch();

						if (batchCounter == executionValue * 10) {
							pstmt.executeBatch();
							pstmt.clearBatch();
							batchCounter = 0;
//							System.out.println(executionValue * 10
//									+ " entries entered successfully");
						}

						batchCounter++;

					}
				}

				if (batchCounter > 0) {
					System.out.println("before executing");
					pstmt.executeBatch();
					System.out.println("after executing");
				}
				// closing
				conn.commit();
				pstmt.close();
				conn.setAutoCommit(true);
			} catch (SQLException e) {
				throw new TaskRunError("Insert Error");
			}
		}
private boolean first = true;
		private void deleteTable(String tableName) throws TaskRunError {
			Statement st = null;
			try {
				st = conn.createStatement();
			} catch (SQLException e1) {
				// TODO Auto-generated catch block
				e1.printStackTrace();
			}
			String sql =
					"Delete From " + tableName;
			try {
				st.executeUpdate(sql);
			} catch (SQLException e) {
				e.printStackTrace();
				throw new TaskRunError("Delete error " + tableName);
			}
		}


		// private function, get year from date
		public String fixDate(String date) {
			int mark = date.indexOf('/');
			String year = null;
			if (mark == -1) {
				mark = date.indexOf('-');
				if (mark != -1) {
					year = date.substring(0, mark);
				}

			} else {
				year = date.substring(date.lastIndexOf('/') + 1, date.length());
			}
			int yearVal = 0;
			try {
				yearVal = Integer.parseInt(year);
			} catch (Exception e) {
				year = "1901";
			}

			if (year.length() != 4) {
				year = "1901";
			}

			if (yearVal < 1901 || yearVal > 2040) {
				year = "1901";
			}

			return year;

		}

		// insert tv program to database
		private void insertProgram(tvPrograms tv_programs) throws TaskRunError {
			PreparedStatement pstmt;
			String firstEpisodeYear = null;
			String finalEpisodeYear = null;
			String numberOfSeasons = null;
			String numberOfEpisodes = null;
			try {
				pstmt = conn
						.prepareStatement("INSERT INTO TV_Program"
								+ "(Name,First_Episode_Date,Final_Episode_Date,Number_of_Episodes,Origin_Country,Number_of_seasons,mid) "
								+ "VALUES(?,?,?,?,?,?,?)");
				conn.setAutoCommit(false);
				for (tvProgram program : tv_programs.getPrograms()) {
					pstmt.setString(1, program.getName());
					firstEpisodeYear = program.getAir_date_of_first_episode();
					firstEpisodeYear = fixDate(firstEpisodeYear);
					pstmt.setString(2, firstEpisodeYear);
					finalEpisodeYear = program.getAir_date_of_first_episode();
					finalEpisodeYear = fixDate(finalEpisodeYear);
					pstmt.setString(3, finalEpisodeYear);
					if (program.getNumber_of_episodes().equals("")) {
						numberOfEpisodes = Integer.toString(0);
					} else {
						numberOfEpisodes = program.getNumber_of_episodes();
					}
					pstmt.setString(4, numberOfEpisodes);
					pstmt.setString(5, program.getCountryOfOrigin());
					if (program.getNumber_of_seasons().equals("")) {
						numberOfSeasons = Integer.toString(0);
					} else {
						numberOfSeasons = program.getNumber_of_seasons();
					}
					pstmt.setString(6, numberOfSeasons);
					pstmt.setString(7, program.getId());
					pstmt.addBatch();
				}
				System.out.println("before executing");
				pstmt.executeBatch();
				System.out.println("after executing");
				// closing
				conn.commit();
				pstmt.close();
				conn.setAutoCommit(true);
			} catch (SQLException e) {
				throw new TaskRunError("Insert Error");
			}
		}

		// insert film to database
		private void insertFilmBatch(Films films) throws TaskRunError {
			PreparedStatement pstmt;
			try {
				pstmt = conn
						.prepareStatement("INSERT INTO Film(Name,Release_Date,Sequel,Prequel,Estimated_Value,Country,mid) VALUES(?,?,?,?,?,?,?)");
				conn.setAutoCommit(false);

				int totalBatch = films.getFilms().size();
				int batchCounter = 0;

				for (Film film : films.getFilms()) {
					pstmt.setString(1, film.getName());
					pstmt.setString(2, fixDate(film.getInitial_release_date()));
					pstmt.setString(3, film.getPrequel());
					pstmt.setString(4, film.getSequel());
					pstmt.setString(5, film.getEstimated_budget());
					pstmt.setString(6, film.getCountry_of_origin());
					pstmt.setString(7, film.getId());
					pstmt.addBatch();

					if (batchCounter == executionValue) {
						pstmt.executeBatch();
						pstmt.clearBatch();
						batchCounter = 0;
					}

					batchCounter++;

				}

				if (batchCounter > 0) {
					pstmt.executeBatch();
				}
				System.out.println("before executing");

				System.out.println("after executing");

				System.out.println("done with film batch!");
				// closing
				conn.commit();
				pstmt.close();
				conn.setAutoCommit(true);
			} catch (SQLException e) {
				e.printStackTrace();
				throw new TaskRunError("Insert Error");
			}
		}

		// insert film play to databae
		private void updateCharacterFilmPlayBatch(List<filmRole> film_roles)
				throws TaskRunError {
			PreparedStatement pstmt;
			int batchCounter = 0;
			try {

				int totalBatch = film_roles.size();
				/*
				pstmt = conn
						.prepareStatement("INSERT INTO Film_Play(`Character`,mid) "
								+ "VALUES(?,?) ON DUPLICATE KEY UPDATE "
								+ "`Character` = VALUES(`Character`), mid = VALUES(mid)");
								*/
				pstmt = conn
						.prepareStatement("UPDATE Film_Play SET `Character` = ? "
								+ "WHERE mid = ? ");				
				conn.setAutoCommit(false);
				for (filmRole role : film_roles) {
					if (role.getCharacter().length() == 0) {
						batchCounter++;
						continue;
					}
					if (role.getId().length() == 0) {
						batchCounter++;
						continue;
					}
					pstmt.setString(1, role.getCharacter());
					pstmt.setString(2, role.getId());
					pstmt.addBatch();

//					System.out.println(executionValue
//							+ " entries entered successfully");
					if (batchCounter == executionValue * 10) {
						batchCounter = 0;
						pstmt.executeBatch();
						pstmt.clearBatch();
//						System.out.println("updated " + executionValue * 10
//								+ "sucssefully!");
					}

					batchCounter++;

				}
				if (batchCounter > 0) {
					System.out.println("before executing");
					pstmt.executeBatch();
					System.out.println("after executing");
				}
				// closing
				conn.commit();
				pstmt.close();
				conn.setAutoCommit(true);
			} catch (SQLException e) {
				e.printStackTrace();
				System.out.println("failed on batch #" + batchCounter);
				throw new TaskRunError("Insert Error");
			}

		}

		// insert directs to database
		private void insertDirectorTableBatch(List<DirectorTable> directorTables)
				throws TaskRunError {
			PreparedStatement pstmt;
			try {
				pstmt = conn
						.prepareStatement("INSERT INTO Directs(Film_Id,Director_Id) VALUES(?,?)");
				conn.setAutoCommit(false);
				for (DirectorTable directorTable : directorTables) {
					if (directorTable.getDirectorId()!=-1 && directorTable.getFilmId()!=-1){
						pstmt.setInt(1, directorTable.getFilmId());
						pstmt.setInt(2, directorTable.getDirectorId());
						pstmt.addBatch();
					}	
				}
				System.out.println("before executing");
				pstmt.executeBatch();
				System.out.println("after executing");
				// closing
				conn.commit();
				pstmt.close();
				conn.setAutoCommit(true);
			} catch (SQLException e) {
				throw new TaskRunError("Insert Error");
			}
		}

		// insert director to database
		private void insertDirectorBatch(filmDirectors filmDirector)
				throws TaskRunError {
			PreparedStatement pstmt;
			try {
				pstmt = conn
						.prepareStatement("INSERT INTO Director(Name,mid) VALUES(?,?)");
				conn.setAutoCommit(false);
				for (filmDirector director : filmDirector.getfilmDirectors()) {
					pstmt.setString(1, director.getName());
					pstmt.setString(2, director.getId());
					pstmt.addBatch();
				}
				System.out.println("before executing");
				pstmt.executeBatch();
				System.out.println("after executing");
				// closing
				conn.commit();
				pstmt.close();
				conn.setAutoCommit(true);
			} catch (SQLException e) {
				e.printStackTrace();
				throw new TaskRunError("Insert Error");
			}
		}

		// insert created table
		private void insertCreatedTable(List<CreatedTable> createdTableList)
				throws TaskRunError {
			PreparedStatement pstmt;
			try {
				pstmt = conn
						.prepareStatement("INSERT INTO Created(Program_Id,Program_Creator_Id) VALUES(?,?)");
				conn.setAutoCommit(false);
				for (CreatedTable table : createdTableList) {
					if (table.getProgramCreatorId() != -1
							&& table.getProgramId() != -1) {
						pstmt.setInt(1, table.getProgramId());
						pstmt.setInt(2, table.getProgramCreatorId());
						pstmt.addBatch();
					}
				}
				System.out.println("before executing");
				pstmt.executeBatch();
				System.out.println("after executing");
				conn.commit();
				// closing
				pstmt.close();
				conn.setAutoCommit(true);
			} catch (SQLException e) {
				e.printStackTrace();
				throw new TaskRunError("Insert Error");
			}
		}

		// insert program creator to database
		private void insertProgramCreator(tvProgramCreators tv_program_creators)
				throws TaskRunError {
			PreparedStatement pstmt;
			try {
				pstmt = conn
						.prepareStatement("INSERT INTO Program_Creator(Name,mid) VALUES(?,?)");
				conn.setAutoCommit(false);
				for (tvProgramCreator creator : tv_program_creators
						.getProgramCreators()) {
					pstmt.setString(1, creator.getName());
					pstmt.setString(2, creator.getId());
					pstmt.addBatch();
				}
				System.out.println("before executing");
				conn.commit();
				pstmt.executeBatch();
				System.out.println("after executing");
				// closing
				pstmt.close();
				conn.setAutoCommit(true);
			} catch (SQLException e) {
				throw new TaskRunError("Insert Error");
			}
		}

		// insert regular appearence to database
		private void insertRegularAppearence(
				Map<String, tvStarringRole> tvStarringRoles)
				throws TaskRunError {

			PreparedStatement pstmt;

			try {
				pstmt = conn
						.prepareStatement("INSERT INTO TV_Play(`Character`,Program_Id,Actor_Id,mid) VALUES(?,?,?,?)");
				conn.setAutoCommit(false);
				Collection<tvStarringRole> values = tvStarringRoles.values();
				for (tvStarringRole starringRole : values) {
					if ((starringRole.getActorId() != -1)
							&& (starringRole.getProgramId() != -1) && starringRole.getId()!=null  && !starringRole.getId().equals("")) {
						pstmt.setString(1, starringRole.getCharacter());
						pstmt.setInt(2, starringRole.getProgramId());
						pstmt.setInt(3, starringRole.getActorId());
						pstmt.setString(4, starringRole.getId());
						pstmt.addBatch();
					}
				}
				System.out.println("before executing");
				pstmt.executeBatch();
				System.out.println("after executing");
				conn.commit();
				// closing
				pstmt.close();
				conn.setAutoCommit(true);
			} catch (SQLException e) {
				e.printStackTrace();
				throw new TaskRunError("Insert Error");
			}
		}

		// insert tv actors to database
		private void insertActorBatch(tvActors tv_actors) throws TaskRunError {
			PreparedStatement pstmt;

			try {
				pstmt = conn
						.prepareStatement("INSERT INTO TV_Actor(Name,mid) VALUES(?,?)");
				conn.setAutoCommit(false);

				int totalBatch = tv_actors.getActors().size();
				int batchCounter = 0;

				for (tvActor actor : tv_actors.getActors()) {
					pstmt.setString(1, actor.getName());
					pstmt.setString(2, actor.getActorID());
					pstmt.addBatch();

					if (batchCounter == executionValue * 10) {
						pstmt.executeBatch();
						batchCounter = 0;
					}

					batchCounter++;
				}
				if (batchCounter > 0) {
					System.out.println("before executing");
					pstmt.executeBatch();
					System.out.println("after executing");
				}
				conn.commit();
				// closing
				pstmt.close();
				conn.setAutoCommit(true);
			} catch (SQLException e) {
				throw new TaskRunError("Insert Error");
			}
		}

		// insert film actors to database
		private void insertFilmActorBatch(filmActors film_actors)
				throws TaskRunError {
			PreparedStatement pstmt;
			try {
				pstmt = conn
						.prepareStatement("INSERT INTO Film_Actor(Name,mid) VALUES(?,?)");
				conn.setAutoCommit(false);
				for (filmActor actor : film_actors.getFilmActors()) {
					pstmt.setString(1, actor.getActor_name());
					pstmt.setString(2, actor.getActor_id());
					pstmt.addBatch();
				}
				System.out.println("before executing");
				pstmt.executeBatch();
				System.out.println("after executing");
				conn.commit();
				// closing
				pstmt.close();
				conn.setAutoCommit(true);
			} catch (SQLException e) {
				throw new TaskRunError("Insert Error");
			}
		}

		

		@Override
		public void updateDataBase(Connection conn) throws TaskRunError {
			// load data types
			this.conn = conn;
			
			// load data types

			// Get a Runtime object
			Runtime r = Runtime.getRuntime();

	//		DebugUtils.initTimer();
	//		DebugUtils.startTimer();

			// Collect garbage at the start of the program

			/* --- tv parsing --- */

			tvActors tv_actors = (tvActors) LoadingCenter
					.loadType(tableType.tvActor);

			System.out.println(tv_actors.getActors().size());
			tvPrograms tv_programs = (tvPrograms) LoadingCenter
					.loadType(tableType.tvProgram);
			System.out.println(tv_programs.getPrograms().size());

			tvProgramCreators tv_program_creators = (tvProgramCreators) LoadingCenter
					.loadType(tableType.tvProgramCreator);
			System.out.println(tv_program_creators.getProgramCreators().size());

			tvStarringRoles starringRoles1 = (tvStarringRoles) LoadingCenter
					.loadType(tableType.tvStarringRole);
			System.out.println(starringRoles1.gettvStarringRoles().size());

			Map<String, Integer> tvCreatorsMap = new HashMap<String, Integer>();
			Map<String, tvStarringRole> tvStarringRoles = new HashMap<String, tvStarringRole>();
			List<tvProgramCreator> programCreators;
			List<CreatedTable> createdTableList = new ArrayList<CreatedTable>();
			int id = 0;

			List<tvStarringRole> tvStarringRolesList;
			System.out.println("here3");
			List<Integer> indexList = IndexRetriever("TV_Actor");
		
			TableOffset offset = new TableOffset(indexList);
			System.out.println("here2");
			// get actor id's and starring roles
			for (tvActor actor : tv_actors.getActors()) {
				id++;
				if(id == 100){
					System.out.println("100");
				}
				id = offset.getID(id);
				tvStarringRolesList = actor.getActorStarringRoles()
						.gettvStarringRoles();
				if (!tvStarringRolesList.isEmpty()) {
					for (tvStarringRole role : tvStarringRolesList) {
						if (role.getId() != null) {
							role.setActorId(id);
							tvStarringRoles.put(role.getId(), role);
						}
					}
				}
			}
			// get tv program creator
			id = 0;
			indexList = IndexRetriever("Program_Creator");
			offset = new TableOffset(indexList);
			for (tvProgramCreator creator : tv_program_creators
					.getProgramCreators()) {
				id++;
				id = offset.getID(id);
				tvCreatorsMap.put(creator.getId(), id);
			}

			// get tv program
			id = 0;
			indexList = IndexRetriever("TV_Program");
			offset = new TableOffset(indexList);
			for (tvProgram program : tv_programs.getPrograms()) {
				id++;
				id = offset.getID(id);
				
				program.setProgramId(id);
				programCreators = program.getProgram_creators()
						.getProgramCreators();
				if (!programCreators.isEmpty()) {
					for (tvProgramCreator creator : programCreators) {
						if (creator.getId() != null
								&& tvCreatorsMap.get(creator.getId()) != null) {
							CreatedTable createdTable = new CreatedTable(id,
									tvCreatorsMap.get(creator.getId()));
							createdTableList.add(createdTable);
						}

					}
				}
				tvStarringRolesList = program.getRegular_cast()
						.gettvStarringRoles();
				if (!tvStarringRolesList.isEmpty()) {
					for (tvStarringRole role : tvStarringRolesList) {
						tvStarringRole tr = tvStarringRoles.get(role.getId());
						if (tr != null) {
							tr.setProgramId(id);
							tvStarringRoles.put(role.getId(), tr);
						}
					}
				}
			}
				System.out.println("here");
		//	 updateActorBatch(tv_actors);
			 updateProgram(tv_programs);
			 updateRegularAppearence(tvStarringRoles);
			 updateProgramCreator(tv_program_creators);
			 deleteTable("Created");
			 setAutoIncremet("Created", 1);
			 insertCreatedTable(createdTableList);
		//	 updateCreatedTable(createdTableList);

			tv_actors = null;
			tv_programs = null;
			tvStarringRoles = null;
			tv_program_creators = null;
			tvCreatorsMap = null;
			tvStarringRoles = null;
			starringRoles1 = null;
			programCreators = null;
			createdTableList = null;
			FB_ParsingUtils.starringRoleMap = null;
			FB_ParsingUtils.programCreatorMap = null;
			System.gc();
			r.gc();
			// get film director
			id = 0;

			List<filmRole> filmRoleList = new ArrayList<filmRole>();
			Map<String, filmRole> filmRoleMap = new TreeMap<String, filmRole>();

			// parse directors
			Map<String, Integer> directorsMap = new HashMap<String, Integer>();
			filmDirectors film_director = (filmDirectors) LoadingCenter
					.loadType(tableType.filmDirector);
			
			indexList = IndexRetriever("Director");
			offset = new TableOffset(indexList);
			
			for (filmDirector director : film_director.getfilmDirectors()) {
				id++;
				id = offset.getID(id);
				director.setFilmDirectorId(id);
				directorsMap.put(director.getId(), id);
			}

			// insert to database and release memory
			updateDirectorBatch(film_director);
			film_director = null;

			/* --- film parsing --- */

			List<DirectorTable> directorTableList = new ArrayList<DirectorTable>();
			Films films = (Films) LoadingCenter.loadType(tableType.film);
			String[] directorsList;
			System.out.println(films.getFilms().size());

			// get films
			id = 0;
			String[] starringRoles;
			
			indexList = IndexRetriever("Film");
			offset = new TableOffset(indexList);
			
			for (Film film : films.getFilms()) {
				id++;
				id = offset.getID(id);
				film.setFilmId(id);
				directorsList = film.getDirectorMids().split(",");
				if (directorsList.length > 0) {
					for (String director : directorsList) {
						if (directorsList != null) {
							if (directorsMap.get(director) != null) {
								DirectorTable directorTable = new DirectorTable(
										directorsMap.get(director), id);
								directorTableList.add(directorTable);
							}
						}
					}
				}

				starringRoles = film.getStarringMids().split(",");

				if (starringRoles.length > 0) {
					for (String role : starringRoles) {
						if (role != null) {
							filmRole fr = new filmRole();
							fr.setFilmId(id);
							fr.setId(role);
							filmRoleMap.put(role, fr);
						}
					}
				}
			}

			// insert film into database and release memory
			updateFilmBatch(films);
			films = null;
			// insert directs table and release memory
			//updateDirectorTableBatch(directorTableList);
			
			deleteTable("Directs");
			setAutoIncremet("Directs", 1);
			insertDirectorTableBatch(directorTableList);
			directorTableList = null;
			directorsMap = null;
			System.gc();

			// parse film actors

			filmActors film_actors = (filmActors) LoadingCenter
					.loadType(tableType.filmActor);
			System.out.println(film_actors.getFilmActors().size());
			
			
			
			id = 0;
			
			indexList = IndexRetriever("Film_Actor");
			offset = new TableOffset(indexList);
			String[] starringRoles3;
			for (filmActor filmActor : film_actors.getFilmActors()) {
				id++;
				id = offset.getID(id);
				filmActor.setFilmActorId(id);
				starringRoles3 = filmActor.getFilmRoleMids().split(",");
				if (starringRoles3.length > 0) {
					for (String role : starringRoles3) {
						if (role != null) {

							filmRole fr = filmRoleMap.get(role);
							if (fr != null) {
								fr.setFilmActorId(id);
							}
						}
					}
				}
			}

		//	System.out.println("befor film actor batch");
			// insert film actors to the table and release memory
			updateFilmActorBatch(film_actors);
			film_actors = null;

			// insert film roles and release memory
			//System.out.println("role size : " + filmRoleMap.size());
			updateFilmStarringRole(filmRoleMap);
			filmRoleMap = null;

			System.gc();

			filmRoles film_roles = (filmRoles) LoadingCenter
					.loadType(tableType.filmRole);
			System.out.println(film_roles.getFilm_roles().size());

			updateCharacterFilmPlayBatch(film_roles.getFilm_roles());
			film_roles = null;

			//System.out.println(DebugUtils.TimeToString(DebugUtils.stopTimer()));
			
			this.conn = null;
		}
		
		public void test2(Connection conn){
			this.conn = conn;
		}
		
		public void test(Connection conn){
			
			try {
				
				Map<String,filmRole> roles = new HashMap<String,filmRole>();
				filmRole f = new filmRole("midtest", "test me2");
				f.setFilmId(2);
				f.setFilmActorId(2);
				roles.put("midtest",f );
				f = new filmRole("midtes2","character");
				f.setFilmActorId(1);
				f.setFilmId(3);
				roles.put("midtest2", f);
				System.out.println("bla    :"+roles.size());
				List<Integer> lst = IndexRetriever("Film_Play");
				TableOffset offset = new TableOffset(lst);
				System.out.println("here");
				updateFilmStarringRole(roles);
				
				
				
				
				
			} catch (TaskRunError e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			
			
			
		}

		private void updateCreatedTable(List<CreatedTable> createdTableList) throws TaskRunError {
			PreparedStatement pstmt;
			try {
				pstmt = conn
						.prepareStatement("INSERT INTO Created(Program_Id,Program_Creator_Id) VALUES(?,?) " +
						"ON DUPLICATE KEY " +
						"UPDATE Program_Id = VALUES(Program_Id), " +
						"Program_Creator_Id = VALUES(Program_Creator_Id)");
				conn.setAutoCommit(false);
				for (CreatedTable table : createdTableList) {
					if (table.getProgramCreatorId() != -1
							&& table.getProgramId() != -1) {
						pstmt.setInt(1, table.getProgramId());
						pstmt.setInt(2, table.getProgramCreatorId());
						pstmt.addBatch();
					}
				}
				System.out.println("before executing");
				pstmt.executeBatch();
				System.out.println("after executing");
				conn.commit();
				// closing
				pstmt.close();
				conn.setAutoCommit(true);
			} catch (SQLException e) {
				throw new TaskRunError("Insert Error");
			}
			
		}

		private void updateFilmActorBatch(filmActors film_actors) throws TaskRunError {
			PreparedStatement pstmt;
			try {
				pstmt = conn
						.prepareStatement("INSERT INTO Film_Actor(Name,mid) VALUES(?,?) " +
						"ON DUPLICATE KEY " +
						"UPDATE Name = VALUES(Name), mid = VALUES(mid)");
				conn.setAutoCommit(false);
				for (filmActor actor : film_actors.getFilmActors()) {
					pstmt.setString(1, actor.getActor_name());
					pstmt.setString(2, actor.getActor_id());
					pstmt.addBatch();
				}
				System.out.println("before executing");
				pstmt.executeBatch();
				System.out.println("after executing");
				conn.commit();
				// closing
				pstmt.close();
				conn.setAutoCommit(true);
			} catch (SQLException e) {
				throw new TaskRunError("Insert Error");
			}
		}

		private void updateDirectorBatch(filmDirectors filmDirector) throws TaskRunError {
			PreparedStatement pstmt;
			try {
				pstmt = conn
						.prepareStatement("INSERT INTO Director(Name,mid) VALUES(?,?) " +
								"ON DUPLICATE KEY " +
								"UPDATE Name = VALUES(Name), mid = VALUES(mid)");
				conn.setAutoCommit(false);
				for (filmDirector director : filmDirector.getfilmDirectors()) {
					pstmt.setString(1, director.getName());
					pstmt.setString(2, director.getId());
					pstmt.addBatch();
				}
				System.out.println("before executing");
				pstmt.executeBatch();
				System.out.println("after executing");
				// closing
				conn.commit();
				pstmt.close();
				conn.setAutoCommit(true);
			} catch (SQLException e) {
				throw new TaskRunError("Insert Error");
			}
			
		}

		private void updateProgramCreator(tvProgramCreators tv_program_creators) throws TaskRunError {
			PreparedStatement pstmt;
			try {
				pstmt = conn
						.prepareStatement("INSERT INTO Program_Creator(Name,mid) VALUES(?,?) " +
						"ON DUPLICATE KEY " +
						"UPDATE Name = VALUES(Name), mid = VALUES(mid)");
				conn.setAutoCommit(false);
				for (tvProgramCreator creator : tv_program_creators
						.getProgramCreators()) {
					pstmt.setString(1, creator.getName());
					pstmt.setString(2, creator.getId());
					pstmt.addBatch();
				}
				System.out.println("before executing");
				conn.commit();
				pstmt.executeBatch();
				System.out.println("after executing");
				// closing
				pstmt.close();
				conn.setAutoCommit(true);
			} catch (SQLException e) {
				throw new TaskRunError("Insert Error");
			}
			
		}

		private void updateRegularAppearence(Map<String, tvStarringRole> tvStarringRoles) throws TaskRunError {
			PreparedStatement pstmt;

			try {
				pstmt = conn
						.prepareStatement("INSERT INTO TV_Play(`Character`,Program_Id,Actor_Id,mid) VALUES(?,?,?,?) " +
						"ON DUPLICATE KEY " +
						"UPDATE `Character` = VALUES(`Character`), " +
						"Program_Id = VALUES(Program_Id), Actor_Id = VALUES(Actor_Id), mid = VALUES(mid)");
				conn.setAutoCommit(false);
				Collection<tvStarringRole> values = tvStarringRoles.values();
				for (tvStarringRole starringRole : values) {
					if(starringRole.getActorId() == 55515 || starringRole.getProgramId() == 101){
						System.out.println("prob");
					}
					
					
					if ((starringRole.getActorId() != -1)
							&& (starringRole.getProgramId() != -1)) {
						pstmt.setString(1, starringRole.getCharacter());
						pstmt.setInt(2, starringRole.getProgramId());
						pstmt.setInt(3, starringRole.getActorId());
						pstmt.setString(4, starringRole.getId());
						pstmt.addBatch();
					}
				}
				System.out.println("before executing");
				pstmt.executeBatch();
				System.out.println("after executing");
				conn.commit();
				// closing
				pstmt.close();
				conn.setAutoCommit(true);
			} catch (SQLException e) {
				e.printStackTrace();
				throw new TaskRunError("Insert Error");
			}
			
		}

		private void updateProgram(tvPrograms tv_programs) throws TaskRunError {
			PreparedStatement pstmt;
			String firstEpisodeYear = null;
			String finalEpisodeYear = null;
			String numberOfSeasons = null;
			String numberOfEpisodes = null;
			try {
			/*	pstmt = conn
						.prepareStatement("INSERT INTO " +
						"TV_Program(Name,First_Episode_Date,Final_Episode_Date,Number_of_Episodes,Origin_Country,Number_of_seasons,mid) " +
						"VALUES(?,?,?,?,?,?,?) " +
						"ON DUPLICATE KEY " +
						"UPDATE Name = VALUES(Name), First_Episode_Date = VALUES(First_Episode_Date), " +
						"Final_Episode_Date = VALUES(Final_Episode_Date), " +
						"Number_of_Episodes = VALUES(Number_of_Episodes), " +
						"Origin_Country = VALUES(Origin_Country), " +
						"Number_of_seasons = VALUES(Number_of_seasons), mid = VALUES(mid)");
						*/
				
				pstmt = conn.prepareStatement("UPDATE TV_Program SET Name = ?, " +
						"First_Episode_Date = ?, " +
						"Final_Episode_Date = ?, " +
						"Number_of_Episodes = ?, " +
						"Origin_Country = ?, " +
						"Number_of_seasons = ? WHERE mid = ?");
				conn.setAutoCommit(false);
				int i = 0;
				int size = midiatorSize("TV_Program");
				for (tvProgram program : tv_programs.getPrograms()) {
					if(i == size)
						break;
					else
						i++;
					pstmt.setString(1, program.getName());
					firstEpisodeYear = program.getAir_date_of_first_episode();
					firstEpisodeYear = fixDate(firstEpisodeYear);
					pstmt.setString(2, firstEpisodeYear);
					finalEpisodeYear = program.getAir_date_of_first_episode();
					finalEpisodeYear = fixDate(finalEpisodeYear);
					pstmt.setString(3, finalEpisodeYear);
					if (program.getNumber_of_episodes().equals("")) {
						numberOfEpisodes = Integer.toString(0);
					} else {
						numberOfEpisodes = program.getNumber_of_episodes();
					}
					pstmt.setString(4, numberOfEpisodes);
					pstmt.setString(5, program.getCountryOfOrigin());
					if (program.getNumber_of_seasons().equals("")) {
						numberOfSeasons = Integer.toString(0);
					} else {
						numberOfSeasons = program.getNumber_of_seasons();
					}
					pstmt.setString(6, numberOfSeasons);
					pstmt.setString(7, program.getId());
					pstmt.addBatch();
					
				}
				System.out.println("before executing");
				setAutoIncremet("TV_Program", 1);
				pstmt.executeBatch();
				System.out.println("after executing");
				
				
				////////////////////////////////////////////////////
				pstmt = conn
						.prepareStatement("INSERT INTO TV_Program"
								+ "(Name,First_Episode_Date,Final_Episode_Date,Number_of_Episodes,Origin_Country,Number_of_seasons,mid) "
								+ "VALUES(?,?,?,?,?,?,?)");
				
				for (tvProgram program : tv_programs.getPrograms().subList(size,(tv_programs.getPrograms().size()-1))) {
					if(i == size)
						break;
					else
						i++;
					pstmt.setString(1, program.getName());
					firstEpisodeYear = program.getAir_date_of_first_episode();
					firstEpisodeYear = fixDate(firstEpisodeYear);
					pstmt.setString(2, firstEpisodeYear);
					finalEpisodeYear = program.getAir_date_of_first_episode();
					finalEpisodeYear = fixDate(finalEpisodeYear);
					pstmt.setString(3, finalEpisodeYear);
					if (program.getNumber_of_episodes().equals("")) {
						numberOfEpisodes = Integer.toString(0);
					} else {
						numberOfEpisodes = program.getNumber_of_episodes();
					}
					pstmt.setString(4, numberOfEpisodes);
					pstmt.setString(5, program.getCountryOfOrigin());
					if (program.getNumber_of_seasons().equals("")) {
						numberOfSeasons = Integer.toString(0);
					} else {
						numberOfSeasons = program.getNumber_of_seasons();
					}
					pstmt.setString(6, numberOfSeasons);
					pstmt.setString(7, program.getId());
					pstmt.addBatch();
					
				}
				System.out.println("before executing");
				pstmt.executeBatch();
				System.out.println("after executing");
				
				
				
				//////////////////////////////////////////////////
				// closing
				conn.commit();
				pstmt.close();
				conn.setAutoCommit(true);
			} catch (SQLException e) {
				e.printStackTrace();
				throw new TaskRunError("Insert Error");
			}
			
		}

		private void updateActorBatch(tvActors tv_actors) throws TaskRunError {
			PreparedStatement pstmt;

			try {
				pstmt = conn
						.prepareStatement("INSERT INTO TV_Actor(Name,mid) VALUES(?,?) " +
								"ON DUPLICATE KEY " +
								"UPDATE Name = VALUES(Name), mid = VALUES(mid)");
				conn.setAutoCommit(false);
				for (tvActor actor : tv_actors.getActors()) {
					pstmt.setString(1, actor.getName());
					pstmt.setString(2, actor.getActorID());
					pstmt.addBatch();
				}
				System.out.println("before executing");
				setAutoIncremet("TV_Actor", 1);
				pstmt.executeBatch();
				System.out.println("after executing");
				conn.commit();
				// closing
				pstmt.close();
				conn.setAutoCommit(true);
			} catch (SQLException e) {
				e.printStackTrace();
				throw new TaskRunError("Insert Error");
			}
			
		}

		private void updateFilmStarringRole(Map<String, filmRole> film_roles) throws TaskRunError {
			PreparedStatement pstmt;
			try {
				pstmt = conn
						.prepareStatement("INSERT INTO Film_Play(Actor_Id,Film_Id,`Character`,mid) "
								+ "VALUES(?,?,?,?) ON DUPLICATE KEY UPDATE " +
								"Actor_Id = VALUES(Actor_Id), Film_Id = VALUES(Film_Id), " +
								"`Character` = VALUES(`Character`), mid = VALUES(mid)");
				conn.setAutoCommit(false);
				Collection<filmRole> values = film_roles.values();

				int batchCounter = 0;

				for (filmRole role : values) {
					if ((role.getFilmActorId() != -1)
							&& (role.getFilmId() != -1)) {

						if (role.getId().length() == 0) {
							batchCounter++;
							continue;
						}

						pstmt.setInt(1, role.getFilmActorId());
						pstmt.setInt(2, role.getFilmId());
						pstmt.setString(3, "");
						pstmt.setString(4, role.getId());
						pstmt.addBatch();

						if (batchCounter == executionValue * 10) {
							pstmt.executeBatch();
							pstmt.clearBatch();
							batchCounter = 0;
//							System.out.println(executionValue * 10
//									+ " entries entered successfully");
						}

						batchCounter++;

					}
				}

				if (batchCounter > 0) {
					System.out.println("before executing");
					pstmt.executeBatch();
					System.out.println("after executing");
				}
				// closing
				conn.commit();
				pstmt.close();
				conn.setAutoCommit(true);
			} catch (SQLException e) {
				throw new TaskRunError("Insert Error");
			}
			
		}

		private void updateDirectorTableBatch(List<DirectorTable> directorTables) throws TaskRunError {
			PreparedStatement pstmt;
			try {
				pstmt = conn
						.prepareStatement("INSERT INTO Directs(Film_Id,Director_Id) VALUES(?,?) " +
								"ON DUPLICATE KEY " +
								"UPDATE Film_Id = VALUES(Film_Id), Director_Id = VALUES(Director_Id)");
				conn.setAutoCommit(false);
				for (DirectorTable directorTable : directorTables) {
					pstmt.setInt(1, directorTable.getFilmId());
					pstmt.setInt(2, directorTable.getDirectorId());
					pstmt.addBatch();
				}
				System.out.println("before executing");
				pstmt.executeBatch();
				System.out.println("after executing");
				// closing
				conn.commit();
				pstmt.close();
				conn.setAutoCommit(true);
			} catch (SQLException e) {
				throw new TaskRunError("Insert Error");
			}
			
		}

		private void updateFilmBatch(Films films) throws TaskRunError {
			PreparedStatement pstmt;
			try {
				pstmt = conn
						.prepareStatement("INSERT INTO " +
								"Film(Name,Release_Date,Sequel,Prequel,Estimated_Value,Country,mid) " +
								"VALUES(?,?,?,?,?,?,?) ON DUPLICATE KEY " +
								"UPDATE Name = VALUES(Name), Release_Date = VALUES(Release_Date), " +
								"Sequel = VALUES(Sequel), Prequel = VALUES(Prequel), " +
								"Estimated_Value = VALUES(Estimated_Value), Country = VALUES(Country), " +
								"mid = VALUES(mid)");
				conn.setAutoCommit(false);
				for (Film film : films.getFilms()) {
					pstmt.setString(1, film.getName());
					pstmt.setString(2, fixDate(film.getInitial_release_date()));
					pstmt.setString(3, film.getPrequel());
					pstmt.setString(4, film.getSequel());
					pstmt.setString(5, film.getEstimated_budget());
					pstmt.setString(6, film.getCountry_of_origin());
					pstmt.setString(7, film.getId());
					pstmt.addBatch();
				}
				System.out.println("before executing");
				pstmt.executeBatch();
				System.out.println("after executing");
				// closing
				conn.commit();
				pstmt.close();
				conn.setAutoCommit(true);
			} catch (SQLException e) {
				throw new TaskRunError("Insert Error");
			}
			
		}
		
		//returns TableOffset list of single insertions to the db
		public List<Integer> IndexRetriever(String table_name) throws TaskRunError{
			Statement stmt;
			ResultSet rs;
			List<Integer> index_list = new ArrayList<Integer>();
			int index=0;
			
			try {
				stmt = conn.createStatement();
				rs = stmt.executeQuery("SELECT "+table_name+".id " +
						"From "+table_name+
						" WHERE LENGTH("+table_name+".mid) > 12 ORDER BY "+table_name+".id");
				
				
					while (rs.next() == true) {
						index = rs.getInt("Id");
						index_list.add(index);
					}
				
				// closing
				rs.close();
				stmt.close();
				return index_list;
			} catch (SQLException e) {
				throw new TaskRunError("offsetRetriever error "+ table_name);
			}
		}
		
		public int midiatorSize(String table_name) throws TaskRunError{
			Statement stmt;
			ResultSet rs;
			
			try {
				stmt = conn.createStatement();
				rs = stmt.executeQuery("SELECT count(*) WHERE LENGTH("+table_name+".mid) < 11 ORDER BY "+table_name+".id");
				int size = 0;
				if(rs.next()){
					size = rs.getInt(1);
				}
				// closing
				rs.close();
				stmt.close();
				return size;
			} catch (SQLException e) {
				throw new TaskRunError("offsetRetriever error "+ table_name);
			}
		}
		
		
		
	}

